package com.ch1.kafka001.constant;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class KafkaInfo {

    public static String HOST="127.0.0.1:9002";
    public static String TOPIC="topic1";


    //生产者配置
    public static Properties producerConfig(
            Class<? extends Serializer> keySer,
            Class<? extends Serializer> valueSer
    ){
       Properties properties = new Properties();
       properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,HOST);
       properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,keySer);
       properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,valueSer);

       return properties;
    }


    //消费者配置
    public static Properties consumerConfig(String groupId,
                                            Class<? extends Deserializer> keyDeserializeClazz,
                                            Class<? extends Deserializer> valueDeserializeClazz){
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,HOST);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,keyDeserializeClazz);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,valueDeserializeClazz);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        return properties;
    }

    public static Map<String,Object> consumerConfigMap(String groupId,
                                                       Class<? extends Deserializer> keyDeserializeClazz,
                                                       Class<? extends Deserializer> valueDeserializeClazz){
        Map<String,Object> properties = new HashMap<String, Object>();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,HOST);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,keyDeserializeClazz);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,valueDeserializeClazz);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        return properties;
    }

    public static void show(ConsumerRecord<String,String> record){
        System.out.println("主题："+record.topic()+",分区："+record.partition()+",偏移量："+
                record.offset()+",key:"+record.key()+",value:"+record.value());
    }
}
